package com.wzh.project.es.client;


import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.http.HttpHost;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.util.EntityUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.client.Request;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;

import com.css.sword.kernel.base.exception.SwordBaseCheckedException;
import com.css.sword.kernel.platform.component.elasticsearch.SwordEsErrorUtils;
import com.css.sword.kernel.platform.component.exception.SwordExceptionParser;
import com.css.sword.utils.SwordJsonUtils;
import com.wzh.project.es.config.ESConfig;
import com.wzh.project.es.http.HttpMethod;

/**
*@author : 汪振海
*@createdDate : Apr 9, 2020  7:34:27 PM
*@description : <br>
*
*/
public class ESClient implements HttpMethod{
	
	private static final Logger logger = LogManager.getLogger(ESClient.class);
	
	private ESConfig esConfig=new ESConfig();
	private RestClient restClient;
	
	private String dateFormat=esConfig.getDateFormat();
	
	//初始化一些信息此处应该从配置文件加载
	public void init(){
		List<String> serverList=new ArrayList<String>();
		serverList.add("127.0.0.1:2181");
		serverList.add("127.0.0.1:2182");
		serverList.add("127.0.0.1:2183");
		esConfig.setServers(serverList);
		restClient=buildClient();
	}
	
	//创建对应的客户端
	private RestClient buildClient() {
		// TODO: 安全相关配置
		
		// 服务器列表主要是获取对应的es集群地址信息
		final List<String> servers = esConfig.getServers();
		final HttpHost[] hosts = new HttpHost[servers.size()];
		for (int i = 0; i < servers.size(); i++) {
			String[] hostAndPort = servers.get(i).trim().split(":");
			hosts[i] = new HttpHost(hostAndPort[0], Integer.valueOf(hostAndPort[1]));
		}
		// 构造客户端
		RestClientBuilder builder = RestClient.builder(hosts);
		
		builder.setMaxRetryTimeoutMillis(this.esConfig.getMaxRetryTimeoutMillis());
		builder.setRequestConfigCallback(
			    new RestClientBuilder.RequestConfigCallback() {
			        @Override
					public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
						return requestConfigBuilder.setSocketTimeout(esConfig.getSocketTimeout()) // 读取超时
								.setConnectTimeout(esConfig.getConnectTimeout()) // 连接超时
								.setConnectionRequestTimeout(esConfig.getConnectionRequestTimeout()); // 连接池获取连接超时
					}
			    });
		builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
			@Override
			public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
				return httpClientBuilder.setMaxConnTotal(esConfig.getMaxConnTotal())	// 最大连接数
								.setMaxConnPerRoute(esConfig.getMaxConnPerRoute());	// 每个路由最大连接数
			}
		});
		return builder.build();
	}
	
	
	public RestClient getRestClient() {
		return this.restClient;
	}
	
	
	private StringBuilder getEndpoint(String index, String id) {
		return new StringBuilder().append("/").append(index).append("/_doc/").append(id);
	}
	// index/_doc
	private StringBuilder getEndpoint(String index) {
		return new StringBuilder().append("/").append(index).append("/_doc");
	}
	
	
	/**
	 * 	执行原生的json操作
	 * @param method   put get post delete 
	 * @param endpoint	索引名字
	 * @param reqJson	原生的es的json请求
	 * @return
	 * @throws SwordBaseCheckedException
	 */
	public String sendJson(String method, String endpoint, String reqJson) throws SwordBaseCheckedException {
		final long startTime = System.currentTimeMillis();
		try {
			Request request = new Request(method, endpoint);
			// 部分请求，如：getDoc,如果id不存在，默认情况EsClientAPI会抛错, 指定ignore后，不抛错，在resJson中found=false
			// 还有update文档不存在时
			request.addParameter("ignore", "404");
			request.setJsonEntity(reqJson);
			Response res = this.restClient.performRequest(request);
			return EntityUtils.toString(res.getEntity());
		} catch (Exception e) {
			StringBuilder msg = new StringBuilder();
			msg.append("sword|es|client|sendjson|error")
				.append("|").append(method)
				.append("|").append(endpoint)
				.append("|").append(reqJson);
			throw new SwordBaseCheckedException(msg.toString(), e);
		} finally {
			logger.debug("当前执行时间"+(System.currentTimeMillis()-startTime));
		}
	}
	
	/**
	 * 保存一个文档，'_id'由ES自动生成
	 * @param index 索引名称 
	 * @param doc 文档
	 * @return _id值
	 * @throws Exception
	 */
	public void putDoc(String index, String _id, Map<String, Object> doc, boolean putIfAbsent) throws Exception {
		if (_id == null || "".equals(_id.trim())) {
			throw new SwordBaseCheckedException("id不能为空");
		}
		final StringBuilder endpoint = this.getEndpoint(index, _id);
		if (putIfAbsent) {//只创建，如果文档存在则报错
			endpoint.append("?op_type=create");
		}
		//final long startTime = Sy;
		try {
			final String reqJson = SwordJsonUtils.toJson(doc, esConfig.getDateFormat());
			final String resJson = this.sendJson(PUT, endpoint.toString(), reqJson);
			Map map = SwordJsonUtils.toObject(resJson, Map.class, esConfig.getDateFormat());
			this.validateError("sword|es|client|putdoc|reserror", map);
		} catch (SwordBaseCheckedException e) {
			throw e;
		} catch (Exception e) {
			throw SwordExceptionParser.processException("sword|es|client|putdoc|error", e);
		} finally {
			//SwordEsTracer.traceOperateEnd(this.dsName, "putDoc", PUT, endpoint.toString(), startTime);
		}
	}
	
	/**
	 * 向指定索引批量保存文档
	 * @param index 索引名称
	 * @param ids _id列表，对应文档的'_id'值
	 * @param docs 文档列表
	 * @param putIfAbsent 参考putDoc接口说明
	 * @return 如果没有错误，返回null，否则返回出现错误的文档信息列表，格式[[id,error], [id,error]]
	 * @throws SwordBaseCheckedException
	 */
	public List<String[]> putDocs(String index, List<String> ids, List<Map<String, Object>> docs, boolean putIfAbsent) throws SwordBaseCheckedException {
		final String endpoint = "/_bulk";
		//final long startTime = SwordEsTracer.traceOperateStart(this.dsName, "putDocs", POST, endpoint, null);
		try {
			final String opType = putIfAbsent ? "create" : "index";
			StringBuilder reqJson = new StringBuilder();
			for (int i = 0; i < docs.size(); i++) {
				Map<String, Object> doc = docs.get(i);
				final String id = ids.get(i);
				reqJson.append("{ \"").append(opType).append("\" : { \"_index\" : \"").append(index).append("\", \"_type\" : \"_doc\", \"_id\":\"").append(id).append("\" } }\n");
				reqJson.append(SwordJsonUtils.toJson(doc, dateFormat));
				reqJson.append("\n");
			}
			String resJson = this.sendJson(POST, endpoint, reqJson.toString());
			// 此处处理与ES的_bulk接口的返回格式有关
			Map resMap = SwordJsonUtils.toObject(resJson, Map.class, dateFormat);
			this.validateError("sword|es|client|putdocs|error", resMap);
			// 有错误
			if (resMap.get("errors").toString().equals("true")) {
				List<Map> items = (List<Map>) resMap.get("items");
				List<String[]> retList = new ArrayList<String[]>(items.size());
				for (Map item : items) {
					Map res = (Map) item.get(opType);
					Map error = (Map) res.get("error");
					if (error != null) {
						retList.add(new String[] {(String) res.get("_id"), (String) error.get("reason")});
					}
				}
				return retList;
			}
			return null;
		} catch (Exception e) {
			throw SwordExceptionParser.processException("sword|es|client|putdocs|error", e);
		} finally {
			//SwordEsTracer.traceOperateEnd(this.dsName, "putDocs", POST, endpoint, startTime);
		}
	}

	/**
	 * 保存一个文档，'_id'由ES自动生成
	 * @param index 索引名称 
	 * @param doc 文档
	 * @return _id值
	 * @throws SwordBaseCheckedException
	 */
	public String postDoc(String index, Map<String, Object> doc) throws SwordBaseCheckedException {
		final StringBuilder endpoint = this.getEndpoint(index);
		//final long startTime = SwordEsTracer.traceOperateStart(this.dsName, "postDoc", POST, endpoint.toString(), null);
		try {
			final String reqJson = SwordJsonUtils.toJson(doc, dateFormat);
			final String resJson = this.sendJson(POST, endpoint.toString(), reqJson);
			Map map = SwordJsonUtils.toObject(resJson, Map.class, dateFormat);
			this.validateError("sword|es|client|postdoc|reserror", map);
			return (String) map.get("_id");
		} catch (SwordBaseCheckedException e) {
			throw e;
		} catch (Exception e) {
			throw SwordExceptionParser.processException("sword|es|client|postdoc|error", e);
		} finally {
			//SwordEsTracer.traceOperateEnd(this.dsName, "postDoc", POST, endpoint.toString(), startTime);
		}
	}
	/**
	 * 尝试处理标准的单个error
	 * @param msg
	 * @param resMap
	 * @throws SwordBaseCheckedException 
	 */
	private void validateError(String msg, Map resMap) throws SwordBaseCheckedException {
		Map error = (Map) resMap.get("error");
		// 出现错误
		if (error != null) {
			throw SwordEsErrorUtils.processError(msg, error);
		}
	}
	
	/**
	 * 批量保存文档,'_id'由ES自动生成
	 * @param index 索引名称
	 * @param docs 文档列表
	 * @return 按docs顺序返回的'_id'值列表
	 * @throws SwordBaseCheckedException
	 */
	public List<String> postDocs(String index, List<Map<String, Object>> docs) throws SwordBaseCheckedException {
		final String endpoint = "/_bulk";
		//final long startTime = SwordEsTracer.traceOperateStart(this.dsName, "postDocs", POST, endpoint, null);
		try {
			StringBuilder reqJson = new StringBuilder();
			for (int i = 0; i < docs.size(); i++) {
				Map<String, Object> doc = docs.get(i);
				reqJson.append("{ \"index\" : { \"_index\" : \"").append(index).append("\", \"_type\" : \"_doc\" } }\n");
				reqJson.append(SwordJsonUtils.toJson(doc, dateFormat));
				reqJson.append("\n");
			}
			String resJson = this.sendJson(POST, endpoint, reqJson.toString());
			Map resMap = SwordJsonUtils.toObject(resJson, Map.class, dateFormat);
			this.validateError("sword|es|client|postdocs|error", resMap);
			// 有错误
			if (resMap.get("errors").toString().equals("true")) {
				throw new SwordBaseCheckedException("sword|es|client|postdocs|error" + resJson);
			}
			// 获取每个id
			final List<String> retIds = new ArrayList<String>(docs.size());
			List<Map<String, Object>> items = (List<Map<String, Object>>) resMap.get("items");
			for (Map<String, Object> item : items) {
				Map<String, Object> res = (Map<String, Object>) item.get("index");
				retIds.add((String) res.get("_id"));
			}
			return retIds;
		} catch (Exception e) {
			throw SwordExceptionParser.processException("sword|es|client|postdocs|error", e);
		} finally {
			//SwordEsTracer.traceOperateEnd(this.dsName, "postDocs", POST, endpoint, startTime);
		}
	}
	/**
	 * 删除文档
	 */
	public void deleteDoc(String index, String id) throws SwordBaseCheckedException {
		final StringBuilder endpoint = this.getEndpoint(index, id);
		//final long startTime = SwordEsTracer.traceOperateStart(this.dsName, "deleteDoc", DELETE, endpoint.toString(), null);
		try {
			final String resJson = this.sendJson(DELETE, endpoint.toString(), null);
			Map map = SwordJsonUtils.toObject(resJson, Map.class, dateFormat);
			this.validateError("sword|es|client|deletedoc|reserror", map);
			// "result": deleted or not_found
		} catch (SwordBaseCheckedException e) {
			throw e;
		} catch (Exception e) {
			throw SwordExceptionParser.processException("sword|es|client|deletedoc|error", e);
		} finally {
			//SwordEsTracer.traceOperateEnd(this.dsName, "deleteDoc", DELETE, endpoint.toString(), startTime);
		}
	}
	/**
	 * 批量删除文档
	 */
	public void deleteDocs(String index, List<String> ids) throws SwordBaseCheckedException {
		final String endpoint = "/_bulk";
		//final long startTime = SwordEsTracer.traceOperateStart(this.dsName, "deleteDocs", POST, endpoint.toString(), null);
		try {
			StringBuilder reqJson = new StringBuilder();
			for (String id : ids) {
				reqJson.append("{ \"delete\" : { \"_index\" : \"").append(index).append("\", \"_type\" : \"_doc\", \"_id\" : \"").append(id).append("\" } }\n");
			}
			String resJson = this.sendJson(POST, endpoint, reqJson.toString());
			Map resMap = SwordJsonUtils.toObject(resJson, Map.class, dateFormat);
			this.validateError("sword|es|client|deletedocs|error", resMap);
			// 有错误
			if (resMap.get("errors").toString().equals("true")) {
				throw new SwordBaseCheckedException("sword|es|client|deletedocs|error" + resJson);
			}
		} catch (Exception e) {
			throw SwordExceptionParser.processException("sword|es|client|deletedocs|error", e);
		} finally {
			//SwordEsTracer.traceOperateEnd(this.dsName, "deleteDocs", POST, endpoint.toString(), startTime);
		}
	}
	/**
	 * 获取文档，返回原始json
	 */
	public String getDoc(String index, String id) throws SwordBaseCheckedException {
		final StringBuilder endpoint = this.getEndpoint(index, id);
		//final long startTime = SwordEsTracer.traceOperateStart(this.dsName, "getDoc", GET, endpoint.toString(), null);
		try {
			return this.sendJson(GET, endpoint.toString(), null);
		} catch (Exception e) {
			throw SwordExceptionParser.processException("sword|es|client|getdoc|error", e);
		} finally {
			//SwordEsTracer.traceOperateEnd(this.dsName, "getDoc", GET, endpoint.toString(), startTime);
		}
	}
	/**
	 * 批量获取文档
	 * @return _mget接口返回的原始json
	 */
	public String getDocs(String index, List<String> ids) throws SwordBaseCheckedException {
		final StringBuilder endpoint = this.getEndpoint(index, "_mget");
		//final long startTime = SwordEsTracer.traceOperateStart(this.dsName, "getDocs", GET, endpoint.toString(), null);
		try {
			// {ids:[id1, id2]}
			Map<String, Object> reqJson = new HashMap<String, Object>();
			reqJson.put("ids", ids);
			return this.sendJson(GET, endpoint.toString(), SwordJsonUtils.toJson(reqJson, dateFormat));
		} catch (Exception e) {
			throw SwordExceptionParser.processException("sword|es|client|getdocs|error", e);
		} finally {
			//SwordEsTracer.traceOperateEnd(this.dsName, "getDocs", GET, endpoint.toString(), startTime);
		}
	}

	/**
	 * 更新文档
	 * @param index 索引名称
	 * @param _id _id值,如果更新的_id不存在文档时，则忽略，不抛错
	 * @param doc 需要更新的field:value
	 * @throws SwordBaseCheckedException
	 */
	public void updateDoc(String index, String id, Map<String, Object> doc) throws SwordBaseCheckedException {
		final StringBuilder endpoint = this.getEndpoint(index, id).append("/_update");
		//final long startTime = SwordEsTracer.traceOperateStart(this.dsName, "updateDoc", POST, endpoint.toString(), null);
		try {
			Map<String, Object> map = new HashMap<String, Object>();
			map.put("doc", doc);
			final String reqJson = SwordJsonUtils.toJson(map, dateFormat);
			String resJson = this.sendJson(POST, endpoint.toString(), reqJson);
			
			Map resMap = SwordJsonUtils.toObject(resJson, Map.class, dateFormat);
			this.validateError("sword|es|client|updatedoc|reserror", resMap);
		} catch (SwordBaseCheckedException e) {
			throw e;
		} catch (Exception e) {
			throw SwordExceptionParser.processException("sword|es|client|updatedoc|error", e);
		} finally {
			//SwordEsTracer.traceOperateEnd(this.dsName, "updateDoc", POST, endpoint.toString(), startTime);
		}
	}
	
	/**
	 *批量 更新文档
	 * @param index 索引名称
	 * @param _id _id值,如果更新的_id不存在文档时，则忽略，不抛错
	 * @param doc 需要更新的field:value
	 * @throws SwordBaseCheckedException
	 */
	public void updateDocs(String index, List<String> ids, List<Map<String, Object>> docs) throws SwordBaseCheckedException {
		final String endpoint = "/_bulk";
		//final long startTime = SwordEsTracer.traceOperateStart(this.dsName, "updateDocs", POST, endpoint, null);
		try {
			StringBuilder reqJson = new StringBuilder();
			for (int i = 0; i < docs.size(); i++) {
				Map<String, Object> doc = docs.get(i);
				String id = ids.get(i);
				reqJson.append("{ \"update\" : { \"_index\" : \"").append(index).append("\", \"_type\" : \"_doc\", \"_id\" : \"").append(id).append("\" } }\n");
				Map<String, Object> tmp = new HashMap<String, Object>();
				tmp.put("doc", doc);
				reqJson.append(SwordJsonUtils.toJson(tmp, dateFormat)).append("\n");
			}
			String resJson = this.sendJson(POST, endpoint, reqJson.toString());
			Map resMap = SwordJsonUtils.toObject(resJson, Map.class, dateFormat);
			this.validateError("sword|es|client|updatedocs|error", resMap);
			// 有错误
			if (resMap.get("errors").toString().equals("true")) {
				throw new SwordBaseCheckedException("sword|es|client|updatedocs|error" + resJson);
			}
		} catch (Exception e) {
			throw SwordExceptionParser.processException("sword|es|client|updatedocs|error", e);
		} finally {
			//SwordEsTracer.traceOperateEnd(this.dsName, "updateDocs", POST, endpoint, startTime);
		}
	}

}
